"""
Case Type   : 基础功能
Case Name   : 以binary形式发布订阅二级分区表增删改数据
Create At   : 2022/11/09
Owner       : opentestcase026
Description :
    1.创建发布
    2.创建订阅
    3.在两个集群创建表
    4.表中插入数据
    5.查询订阅端数据是否同步
    6.发布端修改数据
    7.查询是否同步
    8.删除数据
    9.查询是否同步
    10.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.数据同步
    6.成功
    7.数据同步
    8.成功
    9.数据同步
    10.成功
History     :
"""
import os
import unittest

from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.tb_name = 't_pubsub_0210'
        self.pubname = "pub_pubsub_0210"
        self.subname = "sub_pubsub_0210"
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
                              f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
                              f'-W {self.pri_userdb_sub.db_password}'
        self.wal_level = self.com_pub.show_param("wal_level")
        text = '-----查询集群状态，配置白名单，发布端设置wal_level=logical-----'
        self.log.info(text)
        self.log.info('-----发布端状态-----')
        status = self.commsh_pub.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        self.log.info('-----订阅端状态-----')
        status = self.commsh_sub.get_db_cluster_status(
            'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub.db_host}/32  sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'发布端配置白名单失败:{text}')
        if 'logical' != self.wal_level:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, f'设置wal_level失败:{text}')
            restart_res = self.commsh_pub.restart_db_cluster()
            self.assertTrue(restart_res, f'重启数据库失败:{text}')
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  replication  {self.pri_userdb_pub.db_user}  '
            f'{self.pri_userdb_pub.db_host}/32  sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'订阅端配置白名单失败:{text}')

    def test_pubsub(self):
        text1 = '-----step1:创建发布端;expect:成功-----'
        self.log.info(text1)
        create_pub = f"CREATE PUBLICATION {self.pubname} for all tables;"
        pub_res = self.commsh_pub.execut_db_sql(create_pub,
                                                sql_type=self.user_param_pub)
        self.log.info(pub_res)
        self.assertIn(self.constant.create_pub_succ_msg, pub_res,
                      f'执行失败:{text1}')

        text2 = '-----step2:创建订阅端;expect:成功-----'
        self.log.info(text2)
        guc_res = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, guc_res,
                      f'执行失败:{text2}')
        create_sub = \
            f"create subscription {self.subname} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"publication {self.pubname} with (binary=True);"
        sub_res = self.commsh_sub.execut_db_sql(
            create_sub, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.create_sub_succ_msg,
                      sub_res, f'执行失败:{text2}')

        text3 = '-----step3:两个集群创建表;expect:成功-----'
        self.log.info(text3)
        create_tb = f"""drop table if exists {self.tb_name} cascade;
        create table {self.tb_name}(col_1 int primary key,
        col_2 int NOT NULL, col_3 VARCHAR2(30) NOT NULL) 
        with (FILLFACTOR=80) 
        PARTITION BY list (col_1) SUBPARTITION BY list (col_2)
        (PARTITION p_list_1 VALUES(-1,-2,-3,-4,-5,-6,-7,-8,-9,-10)
         (SUBPARTITION p_list_1_1 VALUES (0,-1,-2,-3,-4,-5,-6,-7,-8,-9),
          SUBPARTITION p_list_1_2 VALUES (default)),
        PARTITION p_list_2 VALUES(0,1,2,3,4,5,6,7,8,9)
         (SUBPARTITION p_list_2_1 VALUES (0,1,2,3,4,5,6,7,8,9),
          SUBPARTITION p_list_2_2 VALUES (default),
          SUBPARTITION p_list_2_3 VALUES (10,11,12,13,14,15,16,17,18,19),
          SUBPARTITION p_list_2_4 VALUES (20,21,22,23,24,25,26,27,28,29),
          SUBPARTITION p_list_2_5 VALUES (30,31,32,33,34,35,36,37,38,39)),
        PARTITION p_list_3 VALUES(10,11,12,13,14,15,16,17,18,19)
         (SUBPARTITION p_list_3_2 VALUES (default)),
        PARTITION p_list_4 VALUES(default),
        PARTITION p_list_5 VALUES(20,21,22,23,24,25,26,27,28,29)
         (SUBPARTITION p_list_5_1 VALUES (0,1,2,3,4,5,6,7,8,9),
          SUBPARTITION p_list_5_2 VALUES (default),
          SUBPARTITION p_list_5_3 VALUES (10,11,12,13,14,15,16,17,18,19),
          SUBPARTITION p_list_5_4 VALUES (20,21,22,23,24,25,26,27,28,29),
          SUBPARTITION p_list_5_5 VALUES (30,31,32,33,34,35,36,37,38,39)),
        PARTITION p_list_6 VALUES(30,31,32,33,34,35,36,37,38,39),
        PARTITION p_list_7 VALUES(40,41,42,43,44,45,46,47,48,49)
         (SUBPARTITION p_list_7_1 VALUES (default))) ENABLE ROW MOVEMENT;"""
        pub_res = self.commsh_pub.execut_db_sql(create_tb,
                                                sql_type=self.user_param_pub)
        self.log.info(pub_res)
        self.assertEqual(pub_res.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, f'发布端建表失败:{text3}')
        sub_res1 = self.commsh_sub.execut_db_sql(
            create_tb, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res1)
        self.assertEqual(sub_res1.count(self.constant.TABLE_CREATE_SUCCESS), 2,
                         f'订阅端建表失败:{text3}')

        text4 = '-----step4:发布端表中插入数据;expect:成功-----'
        self.log.info(text4)
        insert_sql = f"insert into {self.tb_name} " \
                     f"values(2,2,'date'),(13,13,'month'),(11,11,'year');"
        insert_res = self.commsh_pub.execut_db_sql(insert_sql,
                                                   self.user_param_pub)
        self.log.info(insert_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, insert_res,
                      f'执行失败:{text4}')

        text5 = '-----step5:查询订阅端数据;expect:数据同步-----'
        self.log.info(text5)
        res = self.commsh_sub.execut_db_sql(
            f"alter subscription {self.subname} refresh publication;",
            self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertIn(self.constant.alter_sub_succ_msg, res,
                      f"刷新订阅失败:{text5}")
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 20)
        select_sql = f"select * from {self.tb_name};"
        select_res = self.commsh_sub.execut_db_sql(
            select_sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(select_res)
        self.assertIn('2 |     2 | date', select_res, f'执行失败:{text5}')
        self.assertIn('13 |    13 | month', select_res, f'执行失败:{text5}')
        self.assertIn('11 |    11 | year', select_res, f'执行失败:{text5}')

        text6 = '-----step6:修改数据;expect:成功-----'
        self.log.info(text6)
        update_res = self.commsh_pub.execut_db_sql(
            f"update {self.tb_name} set col_1 = 7 where col_1 = 2;",
            self.user_param_pub)
        self.log.info(update_res)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, update_res,
                      f'执行失败:{text6}')

        text7 = '-----step7:查询订阅端数据;expect:数据同步-----'
        self.log.info(text7)
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 20)
        select_res = self.commsh_sub.execut_db_sql(
            select_sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(select_res)
        self.assertIn('7 |     2 | date', select_res, f'执行失败:{text7}')
        self.assertIn('3 rows', select_res, f'执行失败:{text7}')

        text8 = '-----step8:修改数据;expect:成功-----'
        self.log.info(text8)
        delete_res = self.commsh_pub.execut_db_sql(
            f"delete from  {self.tb_name};", self.user_param_pub)
        self.log.info(delete_res)
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, delete_res,
                      f'执行失败:{text8}')

        text9 = '-----step9:查询订阅端数据;expect:数据同步-----'
        self.log.info(text9)
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 20)
        select_res = self.commsh_sub.execut_db_sql(
            select_sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(select_res)
        self.assertIn('0 rows', select_res, f'执行失败:{text9}')

    def tearDown(self):
        text = '-----step10:清理环境 expect: 成功-----'
        self.log.info(text)
        drop_sub_res = self.commsh_sub.execut_db_sql(
            f"drop subscription {self.subname}", self.user_param_sub,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_res)
        drop_pub_res = self.commsh_pub.execut_db_sql(
            f"drop publication {self.pubname};", self.user_param_pub)
        self.log.info(drop_pub_res)
        drop_tb = f"drop table {self.tb_name};"
        drop_tb_res1 = self.commsh_pub.execut_db_sql(
            drop_tb, self.user_param_pub)
        self.log.info(drop_tb_res1)
        drop_tb_res2 = self.commsh_sub.execut_db_sql(
            drop_tb, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_tb_res2)
        wal_level_res = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(wal_level_res)
        pub_guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub.db_host}/32')
        self.log.info(pub_guc_res)
        restart_res = self.commsh_pub.restart_db_cluster()
        sub_guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  replication  {self.pri_userdb_pub.db_user}  '
            f'{self.pri_userdb_pub.db_host}/32', macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_guc_res)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_res,
                      f'删除发布失败:{text}')
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_res,
                      f'删除订阅失败:{text}')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_tb_res1,
                      f'删除订阅端表失败:{text}')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_tb_res2,
                      f'删除发布端表失败:{text}')
        self.assertTrue(wal_level_res, f'执行失败:{text}')
        self.assertTrue(restart_res, f'发布端重启数据库失败:{text}')
        self.assertTrue(pub_guc_res, f'发布端删除白名单失败:{text}')
        self.assertTrue(sub_guc_res, f'订阅端删除白名单失败:{text}')
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
